package com.nageoffer.shortlink.project.mq.consumer;

import com.nageoffer.shortlink.project.common.convention.exception.ServiceException;
import com.nageoffer.shortlink.project.dto.ShortLinkStatsRecodeDTO;
import com.nageoffer.shortlink.project.mq.idempotent.MessageQueueIdempotentHandler;
import com.nageoffer.shortlink.project.service.ShortLinkService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingDeque;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;

import java.util.concurrent.Executors;
import java.util.concurrent.locks.LockSupport;

import static com.nageoffer.shortlink.project.common.constant.RedisKeyConstant.DELAY_QUEUE_STATS_KEY;

/**
 * 延迟记录短链接统计组件
 */
@Slf4j
@Component
@Deprecated
@RequiredArgsConstructor
public class DelayShortLinkStatsConsumer implements InitializingBean {

    private final RedissonClient redissonClient;
    private final ShortLinkService shortLinkService;
    private final MessageQueueIdempotentHandler messageQueueIdempotentHandler;

    public void onMessage(){
        Executors.newSingleThreadExecutor(
                    runnable -> {
                Thread thread = new Thread(runnable);
                thread.setName("delay-short-link_stats_consumer");
                thread.setDaemon(Boolean.TRUE);
                return thread;
        }).execute(() -> {
            RBlockingDeque<ShortLinkStatsRecodeDTO> blockingDeque = redissonClient.getBlockingDeque(DELAY_QUEUE_STATS_KEY);
            RDelayedQueue<ShortLinkStatsRecodeDTO> delayedQueue = redissonClient.getDelayedQueue(blockingDeque);
            for (;;){
                try {
                    ShortLinkStatsRecodeDTO statsRecode = delayedQueue.poll();
                    if (statsRecode != null){
                        if (messageQueueIdempotentHandler.isMessageProcessed(statsRecode.getKeys())) {
                            if (messageQueueIdempotentHandler.isMessageProcessed(statsRecode.getKeys())){
                                if (messageQueueIdempotentHandler.isAccomplish(statsRecode.getKeys())) {
                                    return;
                                }
                                throw new ServiceException("消息未完成流程，需要消息队列重试");
                            }
                        }
                        try {
                            shortLinkService.shortLinkStats(null, statsRecode);
                        } catch (Throwable e) {
                            messageQueueIdempotentHandler.delMessageProcessed(statsRecode.getKeys());
                            log.error("延迟消费短链接监控消费异常");
                        }
                        messageQueueIdempotentHandler.setAccomplish(statsRecode.getKeys());
                        continue;
                    }
                    LockSupport.parkUntil(500);
                }catch (Throwable ignored){
                }
            }
        });
    }

    @Override
    public void afterPropertiesSet() {
        onMessage();
    }
}
